// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_PROTOCOL_STRUCT_HPP_
#define _XBASE_PROTOCOL_STRUCT_HPP_

// 
// 定义协议中用到的结构体类型
// 

#include <stdint.h>
#include <string>
#include <vector>
#include <map>
#include <limits>

#include <boost/shared_ptr.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/buffer.hpp>

#include "constants.hpp" // for tablet_status_ops & command names, IS_IN_RANGE

#include "byteorder.hpp"

namespace xbase
{
	namespace protocol
	{
		struct tablet_info
		{
			uint64_t m_tablet_id;
			std::string m_table_name;
			std::string m_start_row;
			std::string m_end_row;
			uint16_t m_status;

			tablet_info() {}

			tablet_info(uint64_t tablet_id,
				    const std::string &table_name,
				    const std::string &start_row,
				    const std::string &end_row,
				    uint16_t status)
				: m_tablet_id(tablet_id)
				, m_table_name(table_name)
				, m_start_row(start_row)
				, m_end_row(end_row)
				, m_status(status) {}

			// 在cpp中定义，以减少依赖的头文件
			bool in_range(const std::string &row) const {
				return IS_IN_RANGE(row, m_start_row, m_end_row);
			}

			std::string tablet_name() const;
			bool is_root_meta_tablet() const;
			
			// 这里提供一组设置状态的操作

			void status_clear() {
				TabletStatusClear(m_status);
			}
			
			void status_set_online() {
				TabletStatusSetOnline(m_status, true);
			}
			void status_clear_online() {
				TabletStatusSetOnline(m_status, false);
			}
			bool status_is_online() const {
				return TabletStatusGetOnline(m_status);
			}
	
			void status_set_split() {
				TabletStatusSetSplitting(m_status, true);
			}
			void status_clear_split() {
				TabletStatusSetSplitting(m_status, false);
			}
			bool status_is_split() const {
				return TabletStatusGetSplitting(m_status);
			}
	
			void status_set_deleted() {
				TabletStatusSetDeleted(m_status, true);
			}
			void status_clear_deleted() {
				TabletStatusSetDeleted(m_status, false);
			}
			bool status_is_deleted() const {
				return TabletStatusGetDeleted(m_status);
			}

			static inline
			tablet_info get_root_meta_tabletinfo() {
				return tablet_info(0, ROOT_META_TABLET_NAME, "", "", TABLET_STATUS_CLEAR);
			}

			template<typename IArchive>
			bool load(IArchive &in) {
				return in.load(&m_tablet_id) &&
					in.load(&m_table_name) &&
					in.load(&m_start_row) &&
					in.load(&m_end_row) &&
					in.load(&m_status);
			}

			template<typename OArchive>
			bool save(OArchive &out) const {
				return out.save(m_tablet_id) &&
					out.save(m_table_name) &&
					out.save(m_start_row) &&
					out.save(m_end_row) &&
					out.save(m_status);
			}

			std::string to_string() const;
		};

		typedef boost::shared_ptr<tablet_info> tablet_info_ptr;

		struct server_address
		{
			std::string m_hostname;
			uint16_t m_port;
			uint64_t m_startcode;

			server_address() {}

			bool valid() const {
				return (! m_hostname.empty()) &&
					(m_port != 0) &&
					(m_startcode != 0) &&
					((~m_startcode) != 0);
			}

			boost::asio::ip::tcp::endpoint to_endpoint() const {
				return boost::asio::ip::tcp::endpoint(
					boost::asio::ip::address::from_string(m_hostname),
					m_port);
			}

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_hostname) &&
					in.load(&m_port) &&
					in.load(&m_startcode);
			}

			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				return out.save(m_hostname) &&
					out.save(m_port) &&
					out.save(m_startcode);
			}

			std::string to_string() const;
		};

		typedef boost::shared_ptr<server_address> server_address_ptr;

		struct key
		{
			std::string m_row_name;
			std::string m_column_name;
			uint64_t m_timestamp;

			key() {}

			key(const std::string &row,
			    const std::string &column,
			    uint64_t timestamp)
				: m_row_name(row)
				, m_column_name(column)
				, m_timestamp(timestamp) {}

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_row_name) &&
					in.load(&m_column_name) &&
					in.load(&m_timestamp);
			}

			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				return out.save(m_row_name) &&
					out.save(m_column_name) &&
					out.save(m_timestamp);
			}

			std::string to_string () const;
		};

		struct record
		{
			key m_key;
			std::string m_value;

			record() {}

			record(const key &k,
			       const std::string &v)
				: m_key(k)
				, m_value(v) {}

			record(const std::pair<key, std::string> &kv)
				: m_key(kv.first)
				, m_value(kv.second) {}

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_key) &&
					in.load(&m_value);
			}
			
			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				return out.save(m_key) &&
					out.save(m_value);
			}
			
			std::string to_string () const;
		};

		typedef std::vector<record> record_vector;

		inline
		record make_record(const std::string &row,
				   const std::string &column,
				   uint64_t timestamp,
				   const std::string &value) {
			return record( key(row, column, timestamp), value );
		}

		// 将记录按组保存
		// 注意：该对象复制的代价可能较高，最好以指针方式持有
		class row_mutation_data
		{
		public:
			typedef boost::shared_ptr<row_mutation_data> ptr_type;
		
			struct column_data
			{
				uint64_t m_timestamp;
				std::string m_column_name;
				std::string m_value;

				column_data(uint64_t timestamp,
					    const std::string &column_name,
					    const std::string &value)
					: m_timestamp(timestamp)
					, m_column_name(column_name)
					, m_value(value) {}
			};
			typedef std::vector< column_data > columns;
			typedef std::map< std::string, columns > groups; // group name -> columns
			
		public:
			row_mutation_data() {}

			explicit
			row_mutation_data(const std::string &row_name)
				: m_row_name(row_name) {}

			const std::string &row_name() const {
				return m_row_name;
			}
			
			// 改变行名。同样的数据向另一行写时可用该接口
			void change_row_name(const std::string &row_name) {
				assert(! row_name.empty());
				m_row_name = row_name;
			}

			void add(const std::string &group_name,
				 const std::string &column_name,
				 const std::string &value,
				 uint64_t timestamp = NO_TIMESTAMP) {
				columns &c = m_groups[group_name]; // 不存在则会立即插入
				c.push_back(column_data(timestamp, column_name, value));
			}

			void add(const row_mutation_data &data) {
				assert(data.row_name() == this->row_name());
				for(groups::const_iterator it_g = data.m_groups.begin();
				    it_g != data.m_groups.end(); ++it_g)
				{
					columns &c = m_groups[it_g->first];
					c.insert(c.end(),
						 it_g->second.begin(),
						 it_g->second.end());
				}
			}

			std::string to_string() const;

			void fill_record_vector(record_vector &vec,
						uint64_t timestamp) const;

			// 检查是否有数据
			bool empty() const {
				return m_groups.empty();
				// 组是在插入列时创建的，所以有组则必定有数据，因此只需判断groups是否为空即可。
			}

			void clear() {
				m_groups.clear();
			}

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return false;
			}

			// columns中的记录是往后添加的，也就是最新的数
			// 据在最后；而协议中要保证最新的数据放在最前，
			// 所以对columns编码时，要反向遍历！
			template<typename OArchive>
			inline
			bool save(OArchive &out, uint64_t timestamp) const {
				typedef groups::const_iterator iter;
				const uint32_t group_num = static_cast<uint32_t>(m_groups.size());
				bool ok = out.save(group_num);
				const iter end_g = m_groups.end();
				for(iter it_g = m_groups.begin(); ok && (it_g != end_g); ++it_g)
				{
					ok = out.save(it_g->first);
					if(ok)
					{
						uint32_t *bytes_field = reinterpret_cast<uint32_t*>(out.left_buffer());
						ok = out.seek(sizeof(uint32_t), SEEK_CUR);
						if(ok)
						{
							const std::size_t left_bytes = out.left_buffer_size();

							typedef columns::const_reverse_iterator riter;
							const columns &c = it_g->second;
							// const uint32_t num = static_cast<uint32_t>(c.size());
							// ok = out.save(num);
							const riter end_c = c.rend();
							for(riter it_c = c.rbegin(); ok && it_c != end_c; ++it_c)
							{
								ok = out.save(m_row_name) &&
									out.save(it_c->m_column_name) &&
									out.save((it_c->m_timestamp == NO_TIMESTAMP)
										 ? timestamp
										 : it_c->m_timestamp) &&
									out.save(it_c->m_value);
							}
							if(ok)
							{
								const std::size_t bytes =
									left_bytes - out.left_buffer_size();
								assert(bytes <= std::numeric_limits<uint32_t>::max());
								const uint32_t bytes_to_encode = static_cast<uint32_t>(bytes);
								*bytes_field = kb::to_byteorder_cast<typename OArchive::byteorder>(bytes_to_encode);
							}
						}
					}
				}
				return ok;
			}
			
		private:
			std::string m_row_name;
			groups m_groups;
		};

		typedef row_mutation_data::ptr_type row_mutation_data_ptr;

		inline
		bool row_mutation_data_ptr_less(const row_mutation_data_ptr &lhs,
						const row_mutation_data_ptr &rhs)
		{
			return (lhs->row_name() < rhs->row_name());
		}
		
		struct row_mutation_data_for_tabletserver
		{
			typedef boost::asio::const_buffer group_data;
			typedef std::pair<std::string, group_data> group;
			typedef std::vector<group> group_vector;
			
			group_vector m_groups;

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				m_groups.clear();
				uint32_t group_num;
				bool ok = in.load(&group_num);
				for(;ok && (group_num > 0) && (! in.empty()); --group_num)
				{
					std::string name;
					uint32_t bytes = 0;
					ok = in.load(&name) &&
						in.load(&bytes);
					const char * const data_begin = in.left_buffer();
					group_data data(data_begin,
							static_cast<std::size_t>(bytes));
					m_groups.push_back(group(name, data));
					
					ok = ok && in.seek(bytes, SEEK_CUR);
				}
				return ok;
			}
			
			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				return false;
			}

			std::string to_string() const;
		};

		struct group_options
		{
			uint32_t m_compress_method;
			uint32_t m_cache_rate;
			uint32_t m_block_size;
			uint32_t m_max_num_sstable;
			uint32_t m_max_size_sstable;
			uint32_t m_max_size_memtable;

			// 待扩展字段
			uint64_t group_options;
			uint64_t bloom_filter;

			bool valid() const;

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_compress_method) &&
					in.load(&m_cache_rate) &&
					in.load(&m_block_size) &&
					in.load(&m_max_num_sstable) &&
					in.load(&m_max_size_sstable) &&
					in.load(&m_max_size_memtable);
			}
			
			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				return out.save(m_compress_method) &&
					out.save(m_cache_rate) &&
					out.save(m_block_size) &&
					out.save(m_max_num_sstable) &&
					out.save(m_max_size_sstable) &&
					out.save(m_max_size_memtable);
			}

			std::string to_string() const;

			uint32_t calc_max_cache_size() const {
				uint64_t tmp = m_max_size_sstable; // 避免数值溢出
				tmp = tmp * m_cache_rate; // [0,100]
				tmp = tmp / 100;
				return static_cast<uint32_t>(tmp);
			}
		};

		struct group_info
		{
			std::string m_group_name;
			group_options m_options;

			group_info() {}
			
			group_info(const std::string &group_name,
				   const group_options &options)
				: m_group_name(group_name)
				, m_options(options) {}

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_group_name) &&
					in.load(&m_options);
			}
			
			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				return out.save(m_group_name) &&
					out.save(m_options);
			}

			std::string to_string() const;
		};

		typedef boost::shared_ptr<group_info> group_info_ptr;
		typedef std::vector<group_info> group_info_vector;
		typedef boost::shared_ptr<group_info_vector> group_info_vector_ptr;

		struct family_options
		{
			uint32_t m_max_num_version;
			uint32_t m_max_ttl;
			uint32_t m_dead_time;

			// 待扩展字段
			uint64_t family_options;
			uint64_t version_option;

			bool valid() const { return true; }

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_max_num_version) &&
					in.load(&m_max_ttl) &&
					in.load(&m_dead_time);
			}

			template<typename IArchive>
			inline
			bool save(IArchive &out) const {
				return out.save(m_max_num_version) &&
					out.save(m_max_ttl) &&
					out.save(m_dead_time);
			}

			std::string to_string() const;
		};

		struct family_info
		{
			std::string m_family_name;
			std::string m_group_name;
			family_options m_options;

			family_info() {}

			family_info(const std::string &family_name,
				    const std::string &group_name,
				    const family_options &options)
				: m_family_name(family_name)
				, m_group_name(group_name)
				, m_options(options) {}

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_family_name) &&
					in.load(&m_group_name) &&
					in.load(&m_options);
			}

			template<typename IArchive>
			inline
			bool save(IArchive &out) const {
				return out.save(m_family_name) &&
					out.save(m_group_name) &&
					out.save(m_options);
			}

			std::string to_string() const;
		};

		typedef boost::shared_ptr<family_info> family_info_ptr;
		typedef std::vector<family_info> family_info_vector;
		typedef boost::shared_ptr<family_info_vector> family_info_vector_ptr;

		struct table_info
		{
			std::string m_table_name;
			group_info_vector m_groups;
			family_info_vector m_families;

			table_info() {}

			explicit
			table_info(const std::string &table_name)
				: m_table_name(table_name) {}

			// 因为组和列族都很少（最多几百），所以就直接用
			// 最简单的顺序查找；以后若可以保证有序，则用二
			// 分查找；
			
			// 返回列所在的组名及true，若列族不存在，则返回空及false；不检查返回的组是否存在
			const family_info *
			get_family_info(const std::string &family_name) const {
				const family_info_vector::const_iterator it_end = m_families.end();
				for(family_info_vector::const_iterator it = m_families.begin();
				    it != it_end; ++ it)
					if(it->m_family_name == family_name)
						return &(*it);
				return NULL;
			}

			// 返回组信息及true，若组不存在，则返回空组及false
			const group_info *
			get_group_info(const std::string &group_name) const {
				const group_info_vector::const_iterator it_end = m_groups.end();
				for(group_info_vector::const_iterator it = m_groups.begin();
				    it != it_end; ++ it)
					if(it->m_group_name == group_name)
						return &(*it);
				return NULL;
			}
			
			bool has_family(const std::string &family_name) const {
				return get_family_info(family_name) != NULL;
			}

			bool has_group(const std::string &group_name) const {
				return get_group_info(group_name) != NULL;
			}

			bool valid() const;

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_table_name) &&
					in.load(&m_groups) &&
					in.load(&m_families);
			}
			
			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				return out.save(m_table_name) &&
					out.save(m_groups) &&
					out.save(m_families);
			}
	
			std::string to_string() const;
		};

		typedef boost::shared_ptr<table_info> table_info_ptr;

		// 批量写请求中的一行数据
		struct row_data
		{
			record_vector m_records;

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.seek(sizeof(uint32_t), SEEK_CUR) &&
					in.load(&m_records);
			}

			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				OArchive tmp = out;
				bool ok = out.seek(sizeof(uint32_t), SEEK_CUR) &&
					out.save(m_records);
				if(ok)
				{
					const std::size_t bytes = tmp.left_buffer_size() - out.left_buffer_size() - sizeof(uint32_t);
					assert(bytes <= std::numeric_limits<uint32_t>::max());
					const uint32_t bytes_to_encode = static_cast<uint32_t>(bytes);
					tmp.save(bytes_to_encode);
				}
				return ok;
			}

			std::string to_string() const;
		};

		struct row_data_v2
		{
			row_mutation_data_ptr m_mutation_data;
			uint64_t m_timestamp;

			row_data_v2() {}

			row_data_v2(row_mutation_data_ptr ptr,
				    uint64_t timestamp)
				: m_mutation_data(ptr)
				, m_timestamp(timestamp) {}

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return false;
				// return in.seek(sizeof(uint32_t), SEEK_CUR) &&
				// 	in.load(&m_mutation_data);
			}

			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				OArchive tmp = out;
				bool ok = out.seek(sizeof(uint32_t), SEEK_CUR) &&
					m_mutation_data->save(out, m_timestamp);
				if(ok)
				{
					const std::size_t bytes = tmp.left_buffer_size() - out.left_buffer_size() - sizeof(uint32_t);
					assert(bytes <= std::numeric_limits<uint32_t>::max());
					const uint32_t bytes_to_encode = static_cast<uint32_t>(bytes);
					tmp.save(bytes_to_encode);
				}
				return ok;
			}

			std::string to_string() const;
		};
		
		struct row_data_v2_for_tabletserver
		{
			row_mutation_data_for_tabletserver m_mutation_data;
			boost::asio::const_buffer m_raw_data;

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				uint32_t data_bytes;
				bool ok = in.load(&data_bytes);
				if(ok)
				{
					m_raw_data = boost::asio::const_buffer(in.left_buffer(), data_bytes);
					const std::size_t left_buffer_size = in.left_buffer_size();
					ok = in.load(&m_mutation_data);
					const std::size_t left_buffer_size_2 = in.left_buffer_size();
					ok = ok && (static_cast<std::size_t>(data_bytes) == (left_buffer_size - left_buffer_size_2));
				}
				return ok;
			}

			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				OArchive tmp = out;
				bool ok = out.seek(sizeof(uint32_t), SEEK_CUR) &&
					out.save(m_mutation_data);
				if(ok)
				{
					const std::size_t bytes = tmp.left_buffer_size() - out.left_buffer_size() - sizeof(uint32_t);
					assert(bytes <= std::numeric_limits<uint32_t>::max());
					const uint32_t bytes_to_encode = static_cast<uint32_t>(bytes);
					tmp.save(bytes_to_encode);
				}
				return ok;
			}

			std::string to_string() const;
		};
		
		struct server_load
		{
			uint32_t m_total_memory; // 系统总内存量, Byte
			uint32_t m_free_memory;	 // 系统free + buffers + cached 的内存量, Byte
			uint32_t m_cpus_load;	 // cpu负载, /proc/loadavg first field * 100 -> [0, 1000]

			void update_meminfo();
			void update_cpuinfo();

			uint32_t m_input_message_number; // 接收到的消息总数，每次心跳后会重置
			uint32_t m_left_input_message_number; // 当前未处理的已接收的消息数
			uint32_t m_output_message_number; // 发送的消息总数，每次心跳后会重置
			uint32_t m_left_output_message_number; // 未处理的待发送的消息数
			
			uint32_t m_tablet_number;	// 子表数
			uint32_t m_memory_table_size;	// 内存表的大小
			uint32_t m_cached_block_size;	// 缓存的块数据的大小
			uint32_t m_cached_block_number; // 缓存的块的个数

			// constants
			enum
			{
				K = 100 * 1024 * 1024,
				C = 1858,
				A = 20,
				K2 = 10,
				K3 = 10
			};

			uint32_t m_capability; // C, max(cacpability, input_message - left)
			uint32_t m_prev_query_load; // 0, A / 100 * prevcpuload
			uint32_t m_query_load; // prev_query_load + ((input_message / capability) + (left_input / ) )
			uint32_t m_memory_load; // ((memory_table + cached_block) / total_memory) + (K / (total_memory - (1))) //* 10000
			uint32_t m_load;	// memory_load * K3 + queryload

			void update_loadinfo();

			// 重置一些数据
			void reset() {
				m_input_message_number = 0;
				m_left_input_message_number = 0;
				m_output_message_number = 0;
				m_left_output_message_number = 0;
			}

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_total_memory) &&
					in.load(&m_free_memory) &&
					in.load(&m_cpus_load) &&
					in.load(&m_input_message_number) &&
					in.load(&m_left_input_message_number) &&
					in.load(&m_output_message_number) &&
					in.load(&m_left_output_message_number) &&
					in.load(&m_tablet_number) &&
					in.load(&m_memory_table_size) &&
					in.load(&m_cached_block_size) &&
					in.load(&m_cached_block_number) &&
					in.load(&m_capability) &&
					in.load(&m_prev_query_load) &&
					in.load(&m_query_load) &&
					in.load(&m_memory_load) &&
					in.load(&m_load);
			}

			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				return out.save(m_total_memory) &&
					out.save(m_free_memory) &&
					out.save(m_cpus_load) &&
					out.save(m_input_message_number) &&
					out.save(m_left_input_message_number) &&
					out.save(m_output_message_number) &&
					out.save(m_left_output_message_number) &&
					out.save(m_tablet_number) &&
					out.save(m_memory_table_size) &&
					out.save(m_cached_block_size) &&
					out.save(m_cached_block_number) &&
					out.save(m_capability) &&
					out.save(m_prev_query_load) &&
					out.save(m_query_load) &&
					out.save(m_memory_load) &&
					out.save(m_load);
			}

			std::string to_string() const;
		};

		struct tablet_server_info
		{
			server_address m_address;
			server_load m_load;

			template<typename IArchive>
			inline
			bool load(IArchive &in) {
				return in.load(&m_address) &&
					in.load(&m_load);
			}

			template<typename OArchive>
			inline
			bool save(OArchive &out) const {
				return out.save(m_address) &&
					out.save(m_load);
			}

			std::string to_string() const;
		};

	} // namespace protocol

} // namespace xbase

#endif	// _XBASE_PROTOCOL_STRUCT_HPP_
